package com.shujia.map;

import com.shujia.common.Config;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

public class GenderJoinUserMapper extends RichMapFunction<String, Tuple3<String, String, Integer>> {


    private HTableInterface table;

    @Override
    public void open(Configuration parameters) throws Exception {

        //创建hadoop配置文件对象
        //在类名冲突的时候将其中一个改成类全名
        org.apache.hadoop.conf.Configuration configuration = new org.apache.hadoop.conf.Configuration();


        //指定zk地址
        configuration.set("hbase.zookeeper.quorum", Config.getString("hbase.zookeeper.quorum"));
        //建立连接   和zk的连接
        HConnection connection = HConnectionManager.createConnection(configuration);
        table = connection.getTable("user");
    }

    @Override
    public Tuple3<String, String, Integer> map(String value) throws Exception {
        String[] split = value.split("\\|");
        String userId = split[4];
        String sentId = split[1];

        Get get = new Get(userId.getBytes());
        get.addColumn("info".getBytes(), "gender".getBytes());
        Result result = table.get(get);
        String gender = Bytes.toString(result.getValue("info".getBytes(), "gender".getBytes()));

        if (gender == null) {
            gender = "其它";
        }

        return new Tuple3<String, String, Integer>(sentId, gender, 1);
    }
}
